package com.sui.bigdata.sml.web.util;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.constant.CommonConstant;
import com.feidee.fdcommon.util.ToolUtils;
import com.feidee.fdhadoop.kerberos.KerberosAuthentication;
import org.apache.spark.sql.SparkSession;

/**
 * spark工具类
 * @Author caokaizhi
 * @Date 2018-09-15 10:05
 * @Reviewer
 */
public class SparkUtil {
    private static SparkSession localSession;
    private static SparkSession yarnSession;

    public  static SparkSession getLocalSparkSession(){
        if(localSession == null) {
            synchronized (SparkUtil.class) {
                if (localSession == null) {
                    KerberosAuthentication.authenticate();
                    if (CommonConstant.ENV_TEST.equals(CustomConfiguration.getString(CommonConstant.ENV_KEY)) || CommonConstant.ENV_DEV.equals(CustomConfiguration.getString(CommonConstant.ENV_KEY))) {
                        System.setProperty("java.security.auth.login.config", KerberosAuthentication.path + "kafka_spark_jaas.conf");
                    }
                    localSession = SparkSession.builder()
                            .master(CustomConfiguration.getString("spark.master","local"))
                            .appName(CustomConfiguration.getString("spark.app.name","fd-sml"))
                            .getOrCreate();
                }
            }
        }
        return localSession;
    }

    public static SparkSession getYarnSparkSession(){
        KerberosAuthentication.authenticate();
        if (CommonConstant.ENV_TEST.equals(CustomConfiguration.getString(CommonConstant.ENV_KEY)) || CommonConstant.ENV_DEV.equals(CustomConfiguration.getString(CommonConstant.ENV_KEY))) {
            System.setProperty("java.security.auth.login.config", KerberosAuthentication.path + "kafka_spark_jaas.conf");
        }
        System.setProperty("spark.yarn.jars","hdfs://nameservice1/user/spark/spark-2.2.0/*");
        SparkSession yarnSession = SparkSession.builder()
                .master("yarn")
                .appName(CustomConfiguration.getString("spark.app.name","fd-sml-dubbo-provider"))
                .config("spark.yarn.jar", "hdfs://nameservice1/user/spark/spark-assembly-1.6.0-cdh5.7.0-hadoop2.6.0-cdh5.7.0.jar")
                .enableHiveSupport()
                .getOrCreate();
        return yarnSession;
    }

    public synchronized static void close(){
        if(localSession != null){
            localSession.close();
            localSession = null;
        }
    }

    public static String getQueue(String userName){
        String queue = CustomConfiguration.getString("spark."+userName+".queue");
        if(!ToolUtils.isNotNull(queue)){
            queue = "root.risk."+userName;
        }
        return queue;
    }
}
